package com.alibaba.datax.plugin.reader.hdfsreader;

import com.alibaba.datax.common.element.*;
import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.plugin.TaskPluginCollector;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.unstructuredstorage.reader.ColumnEntry;
import com.alibaba.datax.plugin.unstructuredstorage.reader.UnstructuredStorageReaderErrorCode;
import com.alibaba.datax.plugin.unstructuredstorage.reader.UnstructuredStorageReaderUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.io.RCFile;
import org.apache.hadoop.hive.ql.io.RCFileRecordReader;
import org.apache.hadoop.hive.ql.io.orc.OrcFile;
import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat;
import org.apache.hadoop.hive.ql.io.orc.OrcSerde;
import org.apache.hadoop.hive.ql.io.orc.Reader;
import org.apache.hadoop.hive.serde2.columnar.BytesRefArrayWritable;
import org.apache.hadoop.hive.serde2.columnar.BytesRefWritable;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.ReflectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * Created by mingya.wmy on 2015/8/12.
 */
public class DFSUtil {

  private static final Logger LOG = LoggerFactory.getLogger(HdfsReader.Job.class);

  private org.apache.hadoop.conf.Configuration hadoopConf = null;
  private String specifiedFileType = null;
  private Boolean haveKerberos = false;
  private String kerberosKeytabFilePath;
  private String kerberosPrincipal;


  private static final int DIRECTORY_SIZE_GUESS = 16 * 1024;

  public static final String HDFS_DEFAULTFS_KEY = "fs.defaultFS";
  public static final String HADOOP_SECURITY_AUTHENTICATION_KEY = "hadoop.security.authentication";


  /**
   * 将task的cfg配置到hadoop的cfg中
   *
   * @param taskConfig Configuration
   */
  public DFSUtil(Configuration taskConfig) {
    hadoopConf = new org.apache.hadoop.conf.Configuration();
    //io.file.buffer.size 性能参数
    //http://blog.csdn.net/yangjl38/article/details/7583374
    Configuration hdpSiteParams = taskConfig.getConfiguration(Key.HADOOP_CONFIG);
    JSONObject hdpSiteParamsAsJsonObj = JSON.parseObject(taskConfig.getString(Key.HADOOP_CONFIG));
    if (null != hdpSiteParams) {
      Set<String> paramKeys = hdpSiteParams.getKeys();
      for (String each : paramKeys) {
        hadoopConf.set(each, hdpSiteParamsAsJsonObj.getString(each));
      }
    }
    hadoopConf.set(HDFS_DEFAULTFS_KEY, taskConfig.getString(Key.DEFAULT_FS));

    //是否有Kerberos认证
    this.haveKerberos = taskConfig.getBool(Key.HAVE_KERBEROS, false);
    if (haveKerberos) {
      this.kerberosKeytabFilePath = taskConfig.getString(Key.KERBEROS_KEYTAB_FILE_PATH);
      this.kerberosPrincipal = taskConfig.getString(Key.KERBEROS_PRINCIPAL);
      this.hadoopConf.set(HADOOP_SECURITY_AUTHENTICATION_KEY, "kerberos");
    }
    this.kerberosAuthentication(this.kerberosPrincipal, this.kerberosKeytabFilePath);
    LOG.info(String.format("hadoopConfig details:%s", JSON.toJSONString(this.hadoopConf)));
  }

  /**
   * kbs 认证
   *
   * @param kerberosPrincipal      String
   * @param kerberosKeytabFilePath String
   */
  private void kerberosAuthentication(String kerberosPrincipal, String kerberosKeytabFilePath) {
    if (haveKerberos && StringUtils.isNotBlank(this.kerberosPrincipal) && StringUtils
        .isNotBlank(this.kerberosKeytabFilePath)) {
      UserGroupInformation.setConfiguration(this.hadoopConf);
      try {
        UserGroupInformation.loginUserFromKeytab(kerberosPrincipal, kerberosKeytabFilePath);
      } catch (Exception e) {
        String message = String
            .format("kerberos认证失败,请确定kerberosKeytabFilePath[%s]和kerberosPrincipal[%s]填写正确",
                kerberosKeytabFilePath, kerberosPrincipal);
        throw DataXException.asDataXException(HdfsReaderErrorCode.KERBEROS_LOGIN_ERROR, message, e);
      }
    }
  }

  /**
   * 获取指定路径列表下符合条件的所有文件的绝对路径
   *
   * @param srcPaths          路径列表
   * @param specifiedFileType 指定文件类型
   */
  public HashSet<String> getAllFiles(List<String> srcPaths, String specifiedFileType) {

    this.specifiedFileType = specifiedFileType;

    if (!srcPaths.isEmpty()) {
      for (String eachPath : srcPaths) {
        LOG.info(String.format("get HDFS all files in path = [%s]", eachPath));
        getHDFSAllFiles(eachPath);
      }
    }
    return sourceHDFSAllFilesList;
  }

  private HashSet<String> sourceHDFSAllFilesList = new HashSet<String>();

  public HashSet<String> getHDFSAllFiles(String hdfsPath) {

    try {
      FileSystem hdfs = FileSystem.get(hadoopConf);
      //判断hdfsPath是否包含正则符号
      if (hdfsPath.contains("*") || hdfsPath.contains("?")) {
        Path path = new Path(hdfsPath);
        FileStatus stats[] = hdfs.globStatus(path);
        for (FileStatus f : stats) {
          if (f.isFile()) {
            if (f.getLen() == 0) {
              String message = String.format("文件[%s]长度为0，将会跳过不作处理！", hdfsPath);
              LOG.warn(message);
            } else {
              addSourceFileByType(f.getPath().toString());
            }
          } else if (f.isDirectory()) {
            getHDFSAllFilesNORegex(f.getPath().toString(), hdfs);
          }
        }
      } else {
        getHDFSAllFilesNORegex(hdfsPath, hdfs);
      }

      return sourceHDFSAllFilesList;

    } catch (IOException e) {
      String message = String.format("无法读取路径[%s]下的所有文件,请确认您的配置项fs.defaultFS, path的值是否正确，" +
          "是否有读写权限，网络是否已断开！", hdfsPath);
      LOG.error(message);
      throw DataXException.asDataXException(HdfsReaderErrorCode.PATH_CONFIG_ERROR, e);
    }
  }

  private HashSet<String> getHDFSAllFilesNORegex(String path, FileSystem hdfs) throws IOException {

    // 获取要读取的文件的根目录
    Path listFiles = new Path(path);

    // If the network disconnected, this method will retry 45 times
    // each time the retry interval for 20 seconds
    // 获取要读取的文件的根目录的所有二级子文件目录
    FileStatus stats[] = hdfs.listStatus(listFiles);

    for (FileStatus f : stats) {
      // 判断是不是目录，如果是目录，递归调用
      if (f.isDirectory()) {
        LOG.info(String.format("[%s] 是目录, 递归获取该目录下的文件", f.getPath().toString()));
        getHDFSAllFilesNORegex(f.getPath().toString(), hdfs);
      } else if (f.isFile()) {

        addSourceFileByType(f.getPath().toString());
      } else {
        String message = String.format("该路径[%s]文件类型既不是目录也不是文件，插件自动忽略。",
            f.getPath().toString());
        LOG.info(message);
      }
    }
    return sourceHDFSAllFilesList;
  }

  // 根据用户指定的文件类型，将指定的文件类型的路径加入sourceHDFSAllFilesList
  private void addSourceFileByType(String filePath) {
    // 检查file的类型和用户配置的fileType类型是否一致
    boolean isMatchedFileType = checkHdfsFileType(filePath, this.specifiedFileType);

    if (isMatchedFileType) {
      LOG.info(
          String.format("[%s]是[%s]类型的文件, 将该文件加入source files列表", filePath, this.specifiedFileType));
      sourceHDFSAllFilesList.add(filePath);
    } else {
      String message = String.format("文件[%s]的类型与用户配置的fileType类型不一致，" +
              "请确认您配置的目录下面所有文件的类型均为[%s]"
          , filePath, this.specifiedFileType);
      LOG.error(message);
      throw DataXException.asDataXException(
          HdfsReaderErrorCode.FILE_TYPE_UNSUPPORT, message);
    }
  }

  /**
   * 根据输入的hdfs路径，生成 流 返回
   *
   * @param filepath String
   * @return InputStream
   */
  public InputStream getInputStream(String filepath) {
    InputStream inputStream;
    Path path = new Path(filepath);
    try {
      FileSystem fs = FileSystem.get(hadoopConf);
      //If the network disconnected, this method will retry 45 times
      //each time the retry interval for 20 seconds
      inputStream = fs.open(path);
      return inputStream;
    } catch (IOException e) {
      String message = String.format("读取文件 : [%s] 时出错,请确认文件：[%s]存在且配置的用户有权限读取", filepath, filepath);
      throw DataXException.asDataXException(HdfsReaderErrorCode.READ_FILE_ERROR, message, e);
    }
  }

  public void sequenceFileStartRead(String sourceSequenceFilePath, Configuration readerSliceConfig,
      RecordSender recordSender, TaskPluginCollector taskPluginCollector) {
    LOG.info(String.format("Start Read sequence file [%s].", sourceSequenceFilePath));

    Path seqFilePath = new Path(sourceSequenceFilePath);
    SequenceFile.Reader reader = null;
    try {
      //获取SequenceFile.Reader实例
      reader = new SequenceFile.Reader(this.hadoopConf,
          SequenceFile.Reader.file(seqFilePath));
      //获取key 与 value
      Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), this.hadoopConf);
      Text value = new Text();
      while (reader.next(key, value)) {
        if (StringUtils.isNotBlank(value.toString())) {
          UnstructuredStorageReaderUtil.transportOneRecord(recordSender,
              readerSliceConfig, taskPluginCollector, value.toString());
        }
      }
    } catch (Exception e) {
      String message = String.format("SequenceFile.Reader读取文件[%s]时出错", sourceSequenceFilePath);
      LOG.error(message);
      throw DataXException
          .asDataXException(HdfsReaderErrorCode.READ_SEQUENCEFILE_ERROR, message, e);
    } finally {
      IOUtils.closeStream(reader);
      LOG.info("Finally, Close stream SequenceFile.Reader.");
    }

  }

  public void rcFileStartRead(String sourceRcFilePath, Configuration readerSliceConfig,
      RecordSender recordSender, TaskPluginCollector taskPluginCollector) {
    LOG.info(String.format("Start Read rcfile [%s].", sourceRcFilePath));
    List<ColumnEntry> column = UnstructuredStorageReaderUtil
        .getListColumnEntry(readerSliceConfig,
            com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COLUMN);
    // warn: no default value '\N'
    String nullFormat = readerSliceConfig
        .getString(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.NULL_FORMAT);

    Path rcFilePath = new Path(sourceRcFilePath);
    FileSystem fs = null;
    RCFileRecordReader recordReader = null;
    try {
      fs = FileSystem.get(rcFilePath.toUri(), hadoopConf);
      long fileLen = fs.getFileStatus(rcFilePath).getLen();
      FileSplit split = new FileSplit(rcFilePath, 0, fileLen, (String[]) null);
      recordReader = new RCFileRecordReader(hadoopConf, split);
      LongWritable key = new LongWritable();
      BytesRefArrayWritable value = new BytesRefArrayWritable();
      Text txt = new Text();
      while (recordReader.next(key, value)) {
        String[] sourceLine = new String[value.size()];
        txt.clear();
        for (int i = 0; i < value.size(); i++) {
          BytesRefWritable v = value.get(i);
          txt.set(v.getData(), v.getStart(), v.getLength());
          sourceLine[i] = txt.toString();
        }
        UnstructuredStorageReaderUtil.transportOneRecord(recordSender,
            column, sourceLine, nullFormat, taskPluginCollector);
      }

    } catch (IOException e) {
      String message = String.format("读取文件[%s]时出错", sourceRcFilePath);
      LOG.error(message);
      throw DataXException.asDataXException(HdfsReaderErrorCode.READ_RCFILE_ERROR, message, e);
    } finally {
      try {
        if (recordReader != null) {
          recordReader.close();
          LOG.info("Finally, Close RCFileRecordReader.");
        }
      } catch (IOException e) {
        LOG.warn(String.format("finally: 关闭RCFileRecordReader失败, %s", e.getMessage()));
      }
    }

  }

  public void orcFileStartRead(String sourceOrcFilePath, Configuration readerSliceConfig,
      RecordSender recordSender, TaskPluginCollector taskPluginCollector) {
    LOG.info(String.format("Start Read orcfile [%s].", sourceOrcFilePath));
    List<ColumnEntry> column = UnstructuredStorageReaderUtil
        .getListColumnEntry(readerSliceConfig,
            com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COLUMN);
    String nullFormat = readerSliceConfig
        .getString(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.NULL_FORMAT);
    StringBuilder allColumns = new StringBuilder();
    StringBuilder allColumnTypes = new StringBuilder();
    boolean isReadAllColumns = false;
    int columnIndexMax = -1;
    // 判断是否读取所有列
    if (null == column || column.size() == 0) {
      int allColumnsCount = getAllColumnsCount(sourceOrcFilePath);
      columnIndexMax = allColumnsCount - 1;
      isReadAllColumns = true;
    } else {
      columnIndexMax = getMaxIndex(column);
    }
    for (int i = 0; i <= columnIndexMax; i++) {
      allColumns.append("col");
      allColumnTypes.append("string");
      if (i != columnIndexMax) {
        allColumns.append(",");
        allColumnTypes.append(":");
      }
    }
    if (columnIndexMax >= 0) {
      JobConf conf = new JobConf(hadoopConf);
      Path orcFilePath = new Path(sourceOrcFilePath);
      Properties p = new Properties();
      p.setProperty("columns", allColumns.toString());
      p.setProperty("columns.types", allColumnTypes.toString());
      try {
        OrcSerde serde = new OrcSerde();
        serde.initialize(conf, p);
        StructObjectInspector inspector = (StructObjectInspector) serde.getObjectInspector();
        InputFormat<?, ?> in = new OrcInputFormat();
        FileInputFormat.setInputPaths(conf, orcFilePath.toString());

        //If the network disconnected, will retry 45 times, each time the retry interval for 20 seconds
        //Each file as a split
        //TODO multy threads
        InputSplit[] splits = in.getSplits(conf, 1);

        RecordReader reader = in.getRecordReader(splits[0], conf, Reporter.NULL);
        Object key = reader.createKey();
        Object value = reader.createValue();
        // 获取列信息
        List<? extends StructField> fields = inspector.getAllStructFieldRefs();

        List<Object> recordFields;
        while (reader.next(key, value)) {
          recordFields = new ArrayList<Object>();

          for (int i = 0; i <= columnIndexMax; i++) {
            Object field = inspector.getStructFieldData(value, fields.get(i));
            recordFields.add(field);
          }
          transportOneRecord(column, recordFields, recordSender,
              taskPluginCollector, isReadAllColumns, nullFormat);
        }
        reader.close();
      } catch (Exception e) {
        String message = String.format("从orcfile文件路径[%s]中读取数据发生异常，请联系系统管理员。"
            , sourceOrcFilePath);
        LOG.error(message);
        throw DataXException.asDataXException(HdfsReaderErrorCode.READ_FILE_ERROR, message);
      }
    } else {
      String message = String
          .format("请确认您所读取的列配置正确！columnIndexMax 小于0,column:%s", JSON.toJSONString(column));
      throw DataXException.asDataXException(HdfsReaderErrorCode.BAD_CONFIG_VALUE, message);
    }
  }

  private Record transportOneRecord(List<ColumnEntry> columnConfigs, List<Object> recordFields
      , RecordSender recordSender, TaskPluginCollector taskPluginCollector,
      boolean isReadAllColumns, String nullFormat) {
    Record record = recordSender.createRecord();
    Column columnGenerated;
    try {
      if (isReadAllColumns) {
        // 读取所有列，创建都为String类型的column
        for (Object recordField : recordFields) {
          String columnValue = null;
          if (recordField != null) {
            columnValue = recordField.toString();
          }
          columnGenerated = new StringColumn(columnValue);
          record.addColumn(columnGenerated);
        }
      } else {
        for (ColumnEntry columnConfig : columnConfigs) {
          String columnType = columnConfig.getType();
          Integer columnIndex = columnConfig.getIndex();
          String columnConst = columnConfig.getValue();

          String columnValue = null;

          if (null != columnIndex) {
            if (null != recordFields.get(columnIndex)) {
              columnValue = recordFields.get(columnIndex).toString();
            }
          } else {
            columnValue = columnConst;
          }
          Type type = Type.valueOf(columnType.toUpperCase());
          // it's all ok if nullFormat is null
          if (StringUtils.equals(columnValue, nullFormat)) {
            columnValue = null;
          }
          switch (type) {
            case STRING:
              columnGenerated = new StringColumn(columnValue);
              break;
            case LONG:
              try {
                columnGenerated = new LongColumn(columnValue);
              } catch (Exception e) {
                throw new IllegalArgumentException(String.format(
                    "类型转换错误, 无法将[%s] 转换为[%s]", columnValue,
                    "LONG"));
              }
              break;
            case DOUBLE:
              try {
                columnGenerated = new DoubleColumn(columnValue);
              } catch (Exception e) {
                throw new IllegalArgumentException(String.format(
                    "类型转换错误, 无法将[%s] 转换为[%s]", columnValue,
                    "DOUBLE"));
              }
              break;
            case BOOLEAN:
              try {
                columnGenerated = new BoolColumn(columnValue);
              } catch (Exception e) {
                throw new IllegalArgumentException(String.format(
                    "类型转换错误, 无法将[%s] 转换为[%s]", columnValue,
                    "BOOLEAN"));
              }

              break;
            case DATE:
              try {
                if (columnValue == null) {
                  columnGenerated = new DateColumn((Date) null);
                } else {
                  String formatString = columnConfig.getFormat();
                  if (StringUtils.isNotBlank(formatString)) {
                    // 用户自己配置的格式转换
                    SimpleDateFormat format = new SimpleDateFormat(
                        formatString);
                    columnGenerated = new DateColumn(
                        format.parse(columnValue));
                  } else {
                    // 框架尝试转换
                    columnGenerated = new DateColumn(
                        new StringColumn(columnValue)
                            .asDate());
                  }
                }
              } catch (Exception e) {
                throw new IllegalArgumentException(String.format(
                    "类型转换错误, 无法将[%s] 转换为[%s]", columnValue,
                    "DATE"));
              }
              break;
            default:
              String errorMessage = String.format(
                  "您配置的列类型暂不支持 : [%s]", columnType);
              LOG.error(errorMessage);
              throw DataXException
                  .asDataXException(
                      UnstructuredStorageReaderErrorCode.NOT_SUPPORT_TYPE,
                      errorMessage);
          }

          record.addColumn(columnGenerated);
        }
      }
      recordSender.sendToWriter(record);
    } catch (IllegalArgumentException iae) {
      taskPluginCollector
          .collectDirtyRecord(record, iae.getMessage());
    } catch (IndexOutOfBoundsException ioe) {
      taskPluginCollector
          .collectDirtyRecord(record, ioe.getMessage());
    } catch (Exception e) {
      if (e instanceof DataXException) {
        throw (DataXException) e;
      }
      // 每一种转换失败都是脏数据处理,包括数字格式 & 日期格式
      taskPluginCollector.collectDirtyRecord(record, e.getMessage());
    }

    return record;
  }

  private int getAllColumnsCount(String filePath) {
    Path path = new Path(filePath);
    try {
      Reader reader = OrcFile.createReader(path, OrcFile.readerOptions(hadoopConf));
      return reader.getTypes().get(0).getSubtypesCount();
    } catch (IOException e) {
      String message = "读取orcfile column列数失败，请联系系统管理员";
      throw DataXException.asDataXException(HdfsReaderErrorCode.READ_FILE_ERROR, message);
    }
  }

  private int getMaxIndex(List<ColumnEntry> columnConfigs) {
    int maxIndex = -1;
    for (ColumnEntry columnConfig : columnConfigs) {
      Integer columnIndex = columnConfig.getIndex();
      if (columnIndex != null && columnIndex < 0) {
        String message = String.format("您column中配置的index不能小于0，请修改为正确的index,column配置:%s",
            JSON.toJSONString(columnConfigs));
        LOG.error(message);
        throw DataXException
            .asDataXException(HdfsReaderErrorCode.CONFIG_INVALID_EXCEPTION, message);
      } else if (columnIndex != null && columnIndex > maxIndex) {
        maxIndex = columnIndex;
      }
    }
    return maxIndex;
  }

  private enum Type {
    STRING, LONG, BOOLEAN, DOUBLE, DATE,
  }

  public boolean checkHdfsFileType(String filepath, String specifiedFileType) {

    Path file = new Path(filepath);

    try {
      FileSystem fs = FileSystem.get(hadoopConf);
      FSDataInputStream in = fs.open(file);

      if (StringUtils.equalsIgnoreCase(specifiedFileType, Constant.CSV)
          || StringUtils.equalsIgnoreCase(specifiedFileType, Constant.TEXT)) {

        boolean isORC = isORCFile(file, fs, in);// 判断是否是 ORC File
        if (isORC) {
          return false;
        }
        boolean isRC = isRCFile(filepath, in);// 判断是否是 RC File
        if (isRC) {
          return false;
        }
        boolean isSEQ = isSequenceFile(filepath, in);// 判断是否是 Sequence File
        if (isSEQ) {
          return false;
        }
        // 如果不是ORC,RC和SEQ,则默认为是TEXT或CSV类型
        return !isORC && !isRC && !isSEQ;

      } else if (StringUtils.equalsIgnoreCase(specifiedFileType, Constant.ORC)) {

        return isORCFile(file, fs, in);
      } else if (StringUtils.equalsIgnoreCase(specifiedFileType, Constant.RC)) {

        return isRCFile(filepath, in);
      } else if (StringUtils.equalsIgnoreCase(specifiedFileType, Constant.SEQ)) {

        return isSequenceFile(filepath, in);
      }

    } catch (Exception e) {
      String message = String.format("检查文件[%s]类型失败，目前支持ORC,SEQUENCE,RCFile,TEXT,CSV五种格式的文件," +
          "请检查您文件类型和文件是否正确。", filepath);
      LOG.error(message);
      throw DataXException.asDataXException(HdfsReaderErrorCode.READ_FILE_ERROR, message, e);
    }
    return false;
  }

  // 判断file是否是ORC File
  private boolean isORCFile(Path file, FileSystem fs, FSDataInputStream in) {
    try {
      // figure out the size of the file using the option or filesystem
      long size = fs.getFileStatus(file).getLen();

      //read last bytes into buffer to get PostScript
      int readSize = (int) Math.min(size, DIRECTORY_SIZE_GUESS);
      in.seek(size - readSize);
      ByteBuffer buffer = ByteBuffer.allocate(readSize);
      in.readFully(buffer.array(), buffer.arrayOffset() + buffer.position(),
          buffer.remaining());

      //read the PostScript
      //get length of PostScript
      int psLen = buffer.get(readSize - 1) & 0xff;
      int len = OrcFile.MAGIC.length();
      if (psLen < len + 1) {
        return false;
      }
      int offset = buffer.arrayOffset() + buffer.position() + buffer.limit() - 1
          - len;
      byte[] array = buffer.array();
      // now look for the magic string at the end of the postscript.
      if (Text.decode(array, offset, len).equals(OrcFile.MAGIC)) {
        return true;
      } else {
        // If it isn't there, this may be the 0.11.0 version of ORC.
        // Read the first 3 bytes of the file to check for the header
        in.seek(0);
        byte[] header = new byte[len];
        in.readFully(header, 0, len);
        // if it isn't there, this isn't an ORC file
        if (Text.decode(header, 0, len).equals(OrcFile.MAGIC)) {
          return true;
        }
      }
    } catch (IOException e) {
      LOG.info(String.format("检查文件类型: [%s] 不是ORC File.", file.toString()));
    }
    return false;
  }

  // 判断file是否是RC file
  private boolean isRCFile(String filepath, FSDataInputStream in) {

    // The first version of RCFile used the sequence file header.
    final byte[] ORIGINAL_MAGIC = new byte[]{(byte) 'S', (byte) 'E', (byte) 'Q'};
    // The 'magic' bytes at the beginning of the RCFile
    final byte[] RC_MAGIC = new byte[]{(byte) 'R', (byte) 'C', (byte) 'F'};
    // the version that was included with the original magic, which is mapped
    // into ORIGINAL_VERSION
    final byte ORIGINAL_MAGIC_VERSION_WITH_METADATA = 6;
    // All of the versions should be place in this list.
    final int ORIGINAL_VERSION = 0;  // version with SEQ
    final int NEW_MAGIC_VERSION = 1; // version with RCF
    final int CURRENT_VERSION = NEW_MAGIC_VERSION;
    byte version;

    byte[] magic = new byte[RC_MAGIC.length];
    try {
      in.seek(0);
      in.readFully(magic);

      if (Arrays.equals(magic, ORIGINAL_MAGIC)) {
        byte vers = in.readByte();
        if (vers != ORIGINAL_MAGIC_VERSION_WITH_METADATA) {
          return false;
        }
        version = ORIGINAL_VERSION;
      } else {
        if (!Arrays.equals(magic, RC_MAGIC)) {
          return false;
        }

        // Set 'version'
        version = in.readByte();
        if (version > CURRENT_VERSION) {
          return false;
        }
      }

      if (version == ORIGINAL_VERSION) {
        try {
          Class<?> keyCls = hadoopConf.getClassByName(Text.readString(in));
          Class<?> valCls = hadoopConf.getClassByName(Text.readString(in));
          if (!keyCls.equals(RCFile.KeyBuffer.class)
              || !valCls.equals(RCFile.ValueBuffer.class)) {
            return false;
          }
        } catch (ClassNotFoundException e) {
          return false;
        }
      }
      boolean decompress = in.readBoolean(); // is compressed?
      if (version == ORIGINAL_VERSION) {
        // is block-compressed? it should be always false.
        boolean blkCompressed = in.readBoolean();
        if (blkCompressed) {
          return false;
        }
      }
      return true;
    } catch (IOException e) {
      LOG.info(String.format("检查文件类型: [%s] 不是RC File.", filepath));
    }
    return false;
  }

  // 判断file是否是Sequence file
  private boolean isSequenceFile(String filepath, FSDataInputStream in) {
    byte[] SEQ_MAGIC = new byte[]{(byte) 'S', (byte) 'E', (byte) 'Q'};
    byte[] magic = new byte[SEQ_MAGIC.length];
    try {
      in.seek(0);
      in.readFully(magic);
      if (Arrays.equals(magic, SEQ_MAGIC)) {
        return true;
      } else {
        return false;
      }
    } catch (IOException e) {
      LOG.info(String.format("检查文件类型: [%s] 不是Sequence File.", filepath));
    }
    return false;
  }

}
